//package com.tzy.for10evalue.common.kafka;
//
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.slf4j.MDC;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
//import org.springframework.context.ApplicationContext;
//import org.springframework.context.annotation.Primary;
//import org.springframework.kafka.core.KafkaTemplate;
//import org.springframework.kafka.support.KafkaHeaders;
//import org.springframework.messaging.Message;
//import org.springframework.messaging.support.MessageBuilder;
//import org.springframework.stereotype.Service;
//import org.springframework.transaction.annotation.Propagation;
//import org.springframework.transaction.annotation.Transactional;
//import org.springframework.transaction.event.TransactionPhase;
//import org.springframework.transaction.event.TransactionalEventListener;
//import org.springframework.util.StringUtils;
//
//@ConditionalOnClass(value=KafkaTemplate.class)
//@Service
//@Primary
//public class KafkaMessageServiceImpl implements SyncMessageService {
//
//	private static final Logger logger = LoggerFactory.getLogger(KafkaMessageServiceImpl.class);
//
//	@Autowired
//	private KafkaTemplate kafkaTemplate;
//
//	@Autowired
//	private ApplicationContext applicationContext;
//
//	@Value("${spring.application.name:unknown}")
//	private String applicationName ;
//
//
//	/**
//	 * 保证必须有事务
//	 */
//	@Override
//	@Transactional(propagation=Propagation.REQUIRED)
//	public void send(String topic, String message) {
//		MyKafkaEvent m = new MyKafkaEvent(this);
//		m.setTopic(topic);
//		m.setMessage(message);
//		applicationContext.publishEvent(m);
//	}
//
//	/**
//	 * 无事务发送消息
//	 */
//	@Override
//	public void sendWithoutTransaction (String topic, String message) {
//		logger.info("非事务消息:topic[{}],message[{}]",topic,message);
//		this.sendKafka(topic,message);
//	}
//
//    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
//    public void handleKafkaEvent(MyKafkaEvent event) {
//		String topic = event.getTopic();
//		String message = event.getMessage();
//		logger.info("事务消息:topic[{}],message[{}]",topic,message);
//		this.sendKafka(topic,message);
//    }
//
//    private void sendKafka(String topic,String data){
//
//		if(!StringUtils.isEmpty(MDC.get("traceId"))) {
//			Message<String> message = MessageBuilder
//					.withPayload(data)
//					.setHeader(KafkaHeaders.TOPIC, topic)
//					.setHeader("traceId", MDC.get("traceId"))
//					.setHeader("service", applicationName)
//					.build();
//			kafkaTemplate.send(message);
//		}else{
//			kafkaTemplate.send(topic,data);
//		}
//
//	}
//}
